//package com.zdy.ws.ws.consumer;
//
//import com.alibaba.fastjson.JSON;
//import com.alibaba.fastjson.JSONObject;
//import com.allindata.carbontracking.application.constants.KafkaTopicConstant;
//import com.allindata.carbontracking.application.dto.inside.DeviceStatusInsideInput;
//import com.allindata.carbontracking.application.services.equipment.EquAcquisitionEquipmentService;
//import com.allindata.carbontracking.application.services.tio.TioWebSocketService;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.lang3.StringUtils;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.springframework.context.annotation.EnableAspectJAutoProxy;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.kafka.support.Acknowledgment;
//import org.springframework.stereotype.Component;
//
//import javax.annotation.Resource;
//import java.util.List;
//
///**
// * 智光内部协议kafka消费端
// *
// * @author zdy
// */
//@Component
//@EnableAspectJAutoProxy(proxyTargetClass = true, exposeProxy = true)
//@Slf4j
//public class ZhiGuangInsideConsumer {
//
//    @Resource
//    private EquAcquisitionEquipmentService acquisitionEquipmentService;
//    @Resource
//    private TioWebSocketService tioWebSocketService;
//
//    /**
//     * 设备状态改变订阅
//     */
//    @KafkaListener(topics = KafkaTopicConstant.DEVICE_STATUS_TOPIC, groupId = KafkaTopicConstant.DEVICE_STATUS_GROUP)
//    public void consumeDeviceStatusData(ConsumerRecord<String, String> record, Acknowledgment ack) {
//        ack.acknowledge();
//        try {
//            JSONObject json = JSON.parseObject(record.value());
//            List<DeviceStatusInsideInput> deviceList = JSONObject.parseArray(json.getString("dataList"), DeviceStatusInsideInput.class);
//            acquisitionEquipmentService.batchDeviceStatus(deviceList);
//            log.info("【智光内部协议】设备状态kafka消费消息：" + json);
//        } catch (Exception e) {
//            log.error("【智光内部协议】设备状态kafka消费消息异常", e);
//        }
//    }
//
//
//    /**
//     * 设备实时监测订阅
//     */
//    @KafkaListener(topics = KafkaTopicConstant.DEVICE_REAL_TIME_DATA_TOPIC, groupId = KafkaTopicConstant.DEVICE_REAL_TIME_DATA_GROUP)
//    public void consumeDeviceRealTimeData(ConsumerRecord<String, String> record, Acknowledgment ack) {
//        ack.acknowledge();
//        try {
//            JSONObject msgJson = JSON.parseObject(record.value());
//            String deviceId = msgJson.getString("meterId");
//            if (StringUtils.isNotEmpty(deviceId)) {
//                tioWebSocketService.sendMessageByGroup(deviceId, msgJson);
//                log.info("【设备实时监测订阅】kafka消费消息：{}", msgJson);
//            } else {
//                log.error("【设备实时监测订阅】kafka消费消息发送失败：{}", msgJson);
//            }
//        } catch (Exception e) {
//            log.error("【设备实时监测订阅】kafka消费消息异常:{}", e.getMessage());
//        }
//    }
//}
